package com.niit.mobileDevide.datausage;

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;

public class DataUsageReducer extends Reducer<Text, DataUsageBean, Text, Text> {

    @Override
    protected void reduce(Text key, Iterable<DataUsageBean> values, Context context) throws IOException, InterruptedException {
        List<DataUsageBean> highUsageUsers = new ArrayList<>();
        int count = 0;

        for (DataUsageBean val : values) {
            highUsageUsers.add(val);
            count++;
        }

        // 输出符合条件的用户总数
        context.write(new Text("HighDataUsageUserCount"), new Text(String.valueOf(count)));

        if (!highUsageUsers.isEmpty()) {
            // 随机抽取10个用户，最少抽取一个用户
            Collections.shuffle(highUsageUsers);
            List<DataUsageBean> sampledUsers = highUsageUsers.subList(0, Math.min(10, highUsageUsers.size()));

            // 对抽取的用户按数据使用量升序排序
            sampledUsers.sort((u1, u2) -> Double.compare(u1.getDataUsage(), u2.getDataUsage()));

            // 输出排序后的用户ID及其数据使用量
            for (DataUsageBean user : sampledUsers) {
                context.write(new Text("SampledUserId"), new Text(String.format("%d, %.2f", user.getNumAppsInstalled(), user.getDataUsage())));
            }
        }
    }
}